/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.core.redis;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.service.Service;
import com.semidata.rtd.core.tcp.client.FutureCompleteHandler;
import com.semidata.rtd.core.tcp.client.TcpConnection;
import com.semidata.rtd.core.tcp.client.TcpResponseHandler;


/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class RedisConnection implements Service {

	private TcpConnection connection;
	private Queue<RedisCommand> sendingQueue;
	private ReentrantLock lock;
	private DataBuffer dataBuffer;
	private RedisClient client;
	private long sendTime;
	private long timeout;
	private String name;

	public RedisConnection(String name, NioEventLoopGroup group,
	        String hostname, int port, RedisClient client, long timeout) {
		this.client = client;
		this.dataBuffer = new DataBuffer();
		this.sendingQueue = new LinkedList<>();
		this.lock = new ReentrantLock();
		this.timeout = timeout;
		this.name = name;

		this.connection = new TcpConnection(group, hostname, port,
		        new TcpResponseHandler() {

			        @Override
			        public void handle(byte[] data) {
				        dataBuffer.addData(data);

				        try {
					        while (true) {
						        Exception exception = null;
						        Object ret = null;
						        try {
							        ret = RedisProtocol.read(dataBuffer);
							        if (ret == null) {
								        dataBuffer.rollback();
								        break;
							        }
						        } catch (RedisCommandException e) {
							        ret = null;
							        exception = e;
						        }

						        dataBuffer.complete();

						        RedisCommand command;
						        lock.lock();
						        try {
							        command = sendingQueue.poll();
						        } finally {
							        lock.unlock();
						        }
						        command.handle(ret, exception);
					        }
				        } catch (Exception e) {
					        sendTime = 0;
//					        System.out.println("Thread "
//					                + Thread.currentThread().getName()
//					                + " Connection " + getName()
//					                + " reset sendTime");
					        reconnect();
				        }

				        lock.lock();
				        try {
					        if (sendingQueue.isEmpty()) {
						        sendTime = 0;
//						        System.out.println("Thread "
//						                + Thread.currentThread().getName()
//						                + " Connection " + getName()
//						                + " reset sendTime");
						        RedisConnection.this.client.signal();
					        }
				        } finally {
					        lock.unlock();
				        }

			        }

		        });
	}

	public String getName() {
		return name;
	}

	private void reconnect() {
		disconnect();
		lock.lock();
		try {
			while (!sendingQueue.isEmpty()) {
				RedisCommand command = sendingQueue.poll();
				command.handle(null, new RedisTimeoutException());
			}
		} finally {
			lock.unlock();
		}
		connect();
	}

	private void connect() {
		final ReentrantLock lock = new ReentrantLock();
		final Condition cond = lock.newCondition();

		try {
			connection.connect(new FutureCompleteHandler() {

				@Override
				public void handle(ChannelFuture future) {
					lock.lock();
					try {
						cond.signal();
					} finally {
						lock.unlock();
					}
				}
			});
		} catch (Exception e) {
			e.printStackTrace();
		}

		lock.lock();
		try {
			try {
				cond.await();
			} catch (InterruptedException e) {
			}
		} finally {
			lock.unlock();
		}
	}

	private void disconnect() {
		connection.disconnect();
	}

	@Override
	public void start() {
		connect();
	}

	@Override
	public void stop() {
		connection.disconnect();
		sendTime = 0;
		for (RedisCommand command : sendingQueue) {
			command.handle(null, new RedisTimeoutException());
		}
	}

	public boolean isIdel() {
		lock.lock();
		try {
			return sendingQueue.isEmpty();
		} finally {
			lock.unlock();
		}
	}

	public boolean isActive() {
		return connection.isActive();
	}

	public void send(List<RedisCommand> commands) {

		if (!connection.isActive()) {
			for (RedisCommand command : commands) {
				command.handle(null, new RedisTimeoutException("Connection "
				        + getName() + " is not active"));
			}
			return;
		}

		ByteBuf buffer = Unpooled.buffer();
		lock.lock();
		try {
			if (!sendingQueue.isEmpty())
				return;

			for (RedisCommand command : commands) {
				buffer.writeBytes(command.getData());
				sendingQueue.add(command);
			}

			if (sendingQueue.isEmpty())
				return;

		} finally {
			lock.unlock();
		}

		byte[] bytes = new byte[buffer.readableBytes()];
		buffer.readBytes(bytes);
		connection.send(bytes);
		lock.lock();
		try {
			sendTime = System.currentTimeMillis();
//			System.out.println("Thread " + Thread.currentThread().getName()
//			        + " Connection " + getName() + " set sendTime " + sendTime
//			        + " size " + bytes.length);
		} finally {
			lock.unlock();
		}
	}

	public void check() {
		boolean flag = false;
		lock.lock();
		try {
			long now = System.currentTimeMillis();
			if (sendTime != 0 && (now - sendTime) > timeout) {
				flag = true;
			}

			if (!connection.isActive()) {
				flag = true;
			}
		} finally {
			lock.unlock();
		}
		if (flag)
			reconnect();
	}

	public void checkSendingQueue() {
		long now = System.currentTimeMillis();
		lock.lock();
		try {
			for (RedisCommand command : sendingQueue) {
				if (command.getExpireTime() < now) {
					command.handle(null,
					        new RedisTimeoutException("now - expire = "
					                + (now - command.getExpireTime())));
				} else {
					break;
				}
			}
		} finally {
			lock.unlock();
		}
	}

}
